package com.jscloud.bigdata.flink.tableconnectorsinput;

import org.apache.flink.table.api.*;

import static org.apache.flink.table.api.Expressions.$;

/**
 * CREATE DATABASE  `mydb` ;
 * USE `mydb`;
 * DROP TABLE IF EXISTS `my_user`;
 * CREATE TABLE `my_user` (
 * `user_id` int(16)NOT NULL AUTO_INCREMENT,
 * `user_name` varchar(128)DEFAULT NULL,
 * `user_pwd` varchar(128)DEFAULT NULL,
 * `user_addr` varchar(128)DEFAULT NULL,
 * PRIMARY KEY(`user_id`)
 * )ENGINE=InnoDB AUTO_INCREMENT=2 DEFAULT CHARSET=utf8;
 * insert into `my_user`(`user_id`,`user_name`,`user_pwd`,`user_addr`)values(1,'zhangsan','123456','sdfsdf');
 */
public class FlinkTableAPIFromMySQL {
        public static void main(String[] args) {
                //1.创建TableEnvironment
                EnvironmentSettings settings = EnvironmentSettings
                        .newInstance()
                        .build();
                TableEnvironment tEnv = TableEnvironment.create(settings);
                //2.创建 source table
                Schema schema = Schema.newBuilder()
                        .column("user_id", DataTypes.INT())
                        .column("user_name", DataTypes.STRING())
                        .column("user_pwd", DataTypes.STRING())
                        .column("user_addr", DataTypes.STRING())
                        .build();
                tEnv.createTemporaryTable("sourceTable", TableDescriptor.forConnector("jdbc")
                        .schema(schema)
                        .option("url", "jdbc:mysql://bigdata03:3306/mydb?useSSL=false&characterEncoding=utf-8")
                        .option("driver", "com.mysql.jdbc.Driver")
                        .option("table-name", "my_user")
                        .option("username", "root")
                        .option("password", "jsCloud123")
                        .build()
                );
                //3.创建sink table
                tEnv.createTemporaryTable("sinkTable", TableDescriptor.forConnector("print")
                        .schema(schema)
                        .build());

                //4.执行Table API 查询
                Table resultTable = tEnv.from("sourceTable").select($("user_id"), $("user_name"), $("user_pwd"), $("user_addr"));
                //5.输出
                resultTable.executeInsert("sinkTable");
        }
}
